import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;

public class Demo05_state {
    /**
     * 状态：对于每条实时数据都需要基于之前的数据进行汇总，汇总的结果再进行更新，这个就是状态
     *      状态需要落盘存储
     * 键控状态
     *      valueState
     *      ListState
     *      MapState
     *      ReducingState
     *      AggregatingState
     * 算子状态
     *      ListState
     *      UnionState
     *      BroadcastState
     * 累加器
     *      自带累加器
     *      自定义累加器
     */


    /**
     * 1.1 valueState案例
     *     获取截至当前温度的最大值
     * @param sensorDataStream
     */
    public static void keyedState_value(DataStream<Demo02_sensorReading> sensorDataStream){
        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, Double>() {
                            //声明一个value类型的状态值，用于存储截至当前温度的最大值
                            ValueState<Double> maxTempState = null;

                            //开始的时候只执行一次 声明状态
                            @Override
                            public void open(Configuration parameters) throws Exception {
                                //获取一个value类型的状态
                                maxTempState = getRuntimeContext().getState(
                                        new ValueStateDescriptor<Double>("maxTemp",Double.class)
                                );
                            }

                            @Override
                            public Double map(Demo02_sensorReading value) throws Exception {

                                double currentTemp = value.getTemperature();
                                double maxTemp = maxTempState.value()==null?0:maxTempState.value();

                                if(currentTemp>maxTemp){
                                    maxTempState.update(currentTemp);
                                }

                                return maxTempState.value();
                            }
                        }
                ).print();
    }

    /**
     * 1.2 list类型的状态
     *     获取截至当前所有温度值的列表
     * @param sensorDataStream
     */
    public static void keyedState_list(DataStream<Demo02_sensorReading> sensorDataStream){

        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, String>() {

                            ListState<Double> allTempState = null;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                allTempState = getRuntimeContext().getListState(
                                        new ListStateDescriptor<Double>("allTemp",Double.class)
                                );
                            }

                            @Override
                            public String map(Demo02_sensorReading value) throws Exception {

                                double currentTemp = value.getTemperature();
                                allTempState.add(currentTemp);

                                //把另一个列表中的所有元素加到当前状态中
//                                allTempState.addAll();
                                //用另一个列表替换当前的状态
//                                allTempState.update();
                                //清空当前状态列表
//                                allTempState.clear();

                                return allTempState.get().toString();
                            }
                        }
                ).print();
    }


    /**
     * 1.3 map类型的状态
     *     获取截至当前 输出 低温/舒适/高温 每个区间的温度值的数量
     * @param sensorDataStream
     */
    public static void keyedState_map(DataStream<Demo02_sensorReading> sensorDataStream){

        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, String>() {

                            MapState<String,Integer> rangeCountState = null;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                rangeCountState = getRuntimeContext().getMapState(
                                        new MapStateDescriptor<String, Integer>("rangeCount",String.class,Integer.class)
                                );
                            }

                            @Override
                            public String map(Demo02_sensorReading value) throws Exception {

                                //初始化map类型的状态
                                if (rangeCountState.isEmpty()) {
                                    rangeCountState.put("高温区",0);
                                    rangeCountState.put("舒适区",0);
                                    rangeCountState.put("低温区",0);
                                }


                                double currentTemp = value.getTemperature();
                                if(currentTemp>=30){
                                    rangeCountState.put("高温区",rangeCountState.get("高温区")+1);
                                }else if(currentTemp<=10){
                                    rangeCountState.put("低温区",rangeCountState.get("低温区")+1);
                                }else{
                                    rangeCountState.put("舒适区",rangeCountState.get("舒适区")+1);
                                }

                                //获取所有的key
//                                rangeCountState.keys();
                                //把另一个map中的键值对添加到当前状态中
//                                rangeCountState.putAll();
                                //获取所有的value
//                                rangeCountState.values();
                                //获取所有的键值对
//                                rangeCountState.entries();
                                //是否包含某个key
//                                rangeCountState.contains();
                                //移除某个键值对
//                                rangeCountState.remove();

                                return rangeCountState.entries().toString();
                            }
                        }
                ).print();
    }

    /**
     * 1.4 reduce类型的状态
     *     获取截至当前 最新时间戳最大温度值
     * @param sensorDataStream
     */
    public static void keyedState_reduce(DataStream<Demo02_sensorReading> sensorDataStream){

        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, String>() {

                            ReducingState<Demo02_sensorReading> newTSMaxTempState = null;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                newTSMaxTempState = getRuntimeContext().getReducingState(
                                        new ReducingStateDescriptor<Demo02_sensorReading>(
                                                "newTSMaxTemp",
                                                (value1,value2)->
                                                        new Demo02_sensorReading(
                                                                value1.getSensorId(),
                                                                value2.getTimestamp(),
                                                                Math.max(value1.getTemperature(), value2.getTemperature())
                                                        ),
                                                Demo02_sensorReading.class
                                        )
                                );
                            }

                            @Override
                            public String map(Demo02_sensorReading value) throws Exception {

                                newTSMaxTempState.add(value);

                                return newTSMaxTempState.get().toString();
                            }
                        }
                ).print();
    }

    /**
     * 1.5 aggregating类型的状态
     *     获取截至当前 之前所有温度值的平均值
     * @param sensorDataStream
     */
    public static void keyedState_aggregate(DataStream<Demo02_sensorReading> sensorDataStream){

        sensorDataStream
                .keyBy(Demo02_sensorReading::getSensorId)
                .map(
                        new RichMapFunction<Demo02_sensorReading, String>() {

                            AggregatingState<Demo02_sensorReading,Double> avgTempState = null;

                            @Override
                            public void open(Configuration parameters) throws Exception {

                                avgTempState = getRuntimeContext().getAggregatingState(
                                        new AggregatingStateDescriptor<Demo02_sensorReading, Tuple2<Double,Integer>, Double>(
                                                "avgTemp",
                                                new AggregateFunction<Demo02_sensorReading, Tuple2<Double, Integer>, Double>() {
                                                    @Override
                                                    public Tuple2<Double, Integer> createAccumulator() {
                                                        return Tuple2.of(0.0,0);
                                                    }

                                                    @Override
                                                    public Tuple2<Double, Integer> add(Demo02_sensorReading value, Tuple2<Double, Integer> accumulator) {
                                                        return Tuple2.of(accumulator.f0+value.getTemperature(),accumulator.f1+1);
                                                    }

                                                    @Override
                                                    public Double getResult(Tuple2<Double, Integer> accumulator) {
                                                        return accumulator.f0 / accumulator.f1;
                                                    }

                                                    @Override
                                                    public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) {
                                                        return Tuple2.of(a.f0+b.f0,a.f1+b.f1);
                                                    }
                                                },
                                                Types.TUPLE(Types.DOUBLE,Types.INT)
                                        )
                                );
                            }

                            @Override
                            public String map(Demo02_sensorReading value) throws Exception {

                                avgTempState.add(value);

                                return avgTempState.get()+"";
                            }
                        }
                ).print();
    }

    /**
     * 2 广播流
     * 需求：判定实时的车速是否超速
     * @param env
     */
    public static void operationState_broadcast(StreamExecutionEnvironment env){

        //分别用两条流接收卡口摄像头限制速度和实时的汽车通过的速度
        DataStreamSource<String> cameraLimitSpeedSource = env.socketTextStream("node101",8888);
        DataStreamSource<String> carSpeedSource = env.socketTextStream("node101",9999);

        //把字符串类型的卡口信息转成 Demo05_CameraLimitSpeed 类型
        DataStream<Demo05_CameraLimitSpeed> cameraLimitSpeedDataStream = cameraLimitSpeedSource
                .map(line->{
                    String[] fields = line.split(" ");
                    return new Demo05_CameraLimitSpeed(
                            Integer.parseInt(fields[0]),
                            Double.parseDouble(fields[1])
                    );
                });

        //把字符串类型的汽车信息转成 Demo05_CarSpeed 类型
        DataStream<Demo05_CarSpeed> carSpeedDataStream = carSpeedSource
                .map(line->{
                    String[] fields = line.split(" ");
                    return new Demo05_CarSpeed(
                            Integer.parseInt(fields[0]),
                            Integer.parseInt(fields[1]),
                            Double.parseDouble(fields[2])
                    );
                });

        //把卡口信息以算子状态的形式转成广播流
        BroadcastStream<Demo05_CameraLimitSpeed> caremaLimieSpeedBroadcastStream = cameraLimitSpeedDataStream.broadcast(
                new MapStateDescriptor<Integer, Double>(
                        "speedLimit",
                        Integer.class,
                        Double.class
                )
        );

        carSpeedDataStream
                .connect(caremaLimieSpeedBroadcastStream)
                .process(
                        new BroadcastProcessFunction<Demo05_CarSpeed, Demo05_CameraLimitSpeed, String>() {
                            @Override
                            public void processElement(Demo05_CarSpeed value, BroadcastProcessFunction<Demo05_CarSpeed, Demo05_CameraLimitSpeed, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                                //获取广播状态信息
                                ReadOnlyBroadcastState<Integer,Double> cameraLimitSpeedState = ctx.getBroadcastState(
                                        new MapStateDescriptor<Integer, Double>(
                                                "speedLimit",
                                                Integer.class,
                                                Double.class
                                        )
                                );

                                //获取实时的汽车的信息
                                int carId = value.getCarId();
                                int carCameraId = value.getCameraId();
                                double carSpeed = value.getCarSpeed();

                                //设置没有限速信息的卡口，默认限速为 40
                                double limitSpeed = 40;

                                //如果卡口信息的map类型状态中有汽车通过的卡口的话，就有特定的限速数据
                                if (cameraLimitSpeedState.contains(carCameraId)) {
                                    limitSpeed = cameraLimitSpeedState.get(carCameraId);
                                }

                                //如果汽车此时通过该摄像头是的实时速度超过了限速的10%，给警告
                                if(carSpeed >= limitSpeed*1.1){
                                    out.collect(carCameraId+"摄像头限速:"+limitSpeed+";"+carId+"通过速度为："+carSpeed+"已经超速，给予警告");
                                }

                            }

                            @Override
                            public void processBroadcastElement(Demo05_CameraLimitSpeed value, BroadcastProcessFunction<Demo05_CarSpeed, Demo05_CameraLimitSpeed, String>.Context ctx, Collector<String> out) throws Exception {
                                //获取原来的广播状态
                                BroadcastState<Integer,Double> cameraLimitSpeedState = ctx.getBroadcastState(
                                        new MapStateDescriptor<Integer, Double>(
                                                "speedLimit",
                                                Integer.class,
                                                Double.class
                                        )
                                );

                                //此时的cameraId不管原来是否存在，都需要进行添加，如果存在就替换，如果不存在就直接添加
                                cameraLimitSpeedState.put(value.getCameraId(),value.getLimitSpeed());
                            }
                        }
                ).print();
    }

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //接收String类型的原始数据
//        DataStreamSource<String> dataStream = Demo02_source.source_socket(env,9999);
        //把String类型转成SensorReading类型
//        DataStream<Demo02_sensorReading> sensorDataStream = Demo02_transformation.transformation_simple(dataStream);

//        keyedState_value(sensorDataStream);
//        keyedState_list(sensorDataStream);
//        keyedState_map(sensorDataStream);
//        keyedState_reduce(sensorDataStream);
//        keyedState_aggregate(sensorDataStream);
        operationState_broadcast(env);

        try {
            env.execute("state");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
